﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Http;
using System.Text;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace TPLDFAgent
{
    public class SubscriberConfiguration
    {
        public SubscriberConfiguration(string name, string group, string url, string something)
        {

        }
    }
    class ESBMessage
    {
        public string Text { get; set; }
        public ESBMessage(string topic, string text)
        {
            Text = text;
        }
    }
    public sealed class SOABlock : IDisposable,
    IPropagatorBlock<ESBMessage, ESBMessage>,
    IReceivableSourceBlock<ESBMessage>
    {
        private static IReceivableSourceBlock<ESBMessage> innerSourceBlock;

        private static ITargetBlock<ESBMessage> innerTargetBlock;

        //private readonly Party party;

        private bool disposed;

        public SOABlock(string topic)
        //SubscriberConfiguration clientConfig)//, SubscriberOptions options)
        {
            //this.party = new Party(clientConfig, options);
            var bufferBlock = new BufferBlock<ESBMessage>();
            SOABlock.innerSourceBlock = bufferBlock;
            var actionBlock = new ActionBlock<ESBMessage>(async
                message =>
                {
                    HttpClient client = new HttpClient();
                    string msg = await client.GetStringAsync("http://localhost:3001/hola/" + message.Text);
                    ESBMessage esbMsg = new ESBMessage(topic, msg);
                    bufferBlock.Post(esbMsg);
                    //.ContinueWith((t) =>
                    //     {
                    //if (t.Exception != null)
                    //{
                    //    this.Fault(t.Exception);

                    //}
                    //else
                    //{
                    //    string msg = t.Result;
                    //    ESBMessage esbMsg = new ESBMessage(topic, msg);
                    //    bufferBlock.Post(esbMsg); 
                    //}
                    // }
                    //);

                }, new ExecutionDataflowBlockOptions{MaxDegreeOfParallelism=10});
            SOABlock.innerTargetBlock = actionBlock;

            actionBlock.Completion.ContinueWith(
                t =>
                {
                    if (t.IsFaulted)
                    {
                        ((IDataflowBlock)bufferBlock).Fault(t.Exception);
                    }
                    else
                    {
                        bufferBlock.Complete();
                    }
                });
            //this.party.OnReceive += (sender, e) =>
            //{
            //    if (null != e.Exception)
            //    {
            //        this.Fault(e.Exception);
            //    }
            //    else
            //    {
            //        bufferBlock.Post(e.Message);
            //    }
            //};
        }

        ~SOABlock()
        {
            this.Dispose(false);
        }

        public Task Completion
        {
            get
            {
                return SOABlock.innerSourceBlock.Completion;
            }
        }

        public void Complete()
        {
            SOABlock.innerTargetBlock.Complete();
        }

        public void Connect()
        {
            //this.party.Connect();
        }

        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        public void Fault(Exception exception)
        {
            SOABlock.innerTargetBlock.Fault(exception);
        }

        bool IReceivableSourceBlock<ESBMessage>.TryReceive(
            Predicate<ESBMessage> filter, out ESBMessage item)
        {
            return SOABlock.innerSourceBlock.TryReceive(filter, out item);
        }

        bool IReceivableSourceBlock<ESBMessage>.TryReceiveAll(out IList<ESBMessage> items)
        {
            return SOABlock.innerSourceBlock.TryReceiveAll(out items);
        }

        ESBMessage ISourceBlock<ESBMessage>.ConsumeMessage(
            DataflowMessageHeader messageHeader, ITargetBlock<ESBMessage> target,
            out bool messageConsumed)
        {
            return SOABlock.innerSourceBlock.ConsumeMessage(
                messageHeader, target, out messageConsumed);
        }

        IDisposable ISourceBlock<ESBMessage>.LinkTo(
            ITargetBlock<ESBMessage> target, DataflowLinkOptions linkOptions)
        {
            return SOABlock.innerSourceBlock.LinkTo(target, linkOptions);
        }

        void ISourceBlock<ESBMessage>.ReleaseReservation(
            DataflowMessageHeader messageHeader, ITargetBlock<ESBMessage> target)
        {
            SOABlock.innerSourceBlock.ReleaseReservation(messageHeader, target);
        }

        bool ISourceBlock<ESBMessage>.ReserveMessage(
            DataflowMessageHeader messageHeader, ITargetBlock<ESBMessage> target)
        {
            return SOABlock.innerSourceBlock.ReserveMessage(messageHeader, target);
        }

        DataflowMessageStatus ITargetBlock<ESBMessage>.OfferMessage(
            DataflowMessageHeader messageHeader,
            ESBMessage messageValue,
            ISourceBlock<ESBMessage> source,
            bool consumeToAccept)
        {
            return SOABlock.innerTargetBlock.OfferMessage(
                messageHeader, messageValue, source, consumeToAccept);
        }

        private void Dispose(bool disposing)
        {
            if (this.disposed)
            {
                return;
            }

            //this.party.Dispose();

            if (!disposing)
            {
                return;
            }

            this.disposed = true;
        }
    }
}
